package cn.com.gf.cmp.am.monitor.missing;

import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import javax.annotation.Resource;

import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import com.fasterxml.jackson.databind.ObjectMapper;

import cn.com.gf.cmp.am.message.MessageTransform;
import cn.com.gf.cmp.am.message.RequestMessage;
import cn.com.gf.cmp.am.message.ResponseMessage;
 

/*
 * 发现oracle入库不按序列号导致数据丢失
 * 使用15分钟一次补取遗漏数据
 * @author zzh
 * 
 * */
@Component
public class MissingSourceMonitor {

   private final Logger logger = LoggerFactory.getLogger(MissingSourceMonitor.class);
	
	@Resource
	private JdbcTemplate jdbcTemplate;
	
	@Resource(name="tradeKafkaTemplate")
	private KafkaTemplate<String, String> template;
	
	@Value("${spring.kafka.amentrust.topics}")
	private String entrustTopic;

	@Value("${spring.kafka.amtrade.topics}")
	private String tradeTopic;
	
	@Value("${table.trade}")
	private String tradeTable = "trealdeal";

	@Value("${table.trade.kafka}")
	private String tradeKafkaTable="trealdeal_kafka";
	
	@Value("${table.entrust}")
	private String entrustTable = "tentrusts";

	@Value("${table.entrust.kafka}")
	private String entrustKafkaTable="tentrusts_kafka";
	
	 
	
    private String SELECT_SQL_ENTRUST ;
    private String SELECT_SQL_TRADE  ;
	private String INSERT_SQL_ENTRUST ;
	private String INSERT_SQL_TRADE;
	
	private MissingSourceContext context=new MissingSourceContext();
	
	
	private ObjectMapper mapper = new ObjectMapper();
	
	@Scheduled(fixedDelay=20*60*1000)
	public void loadAndProduce() {
		try{
			
			//第一次执行时不运行
			if(context.getRuningTime()==0){
				context.setRuningTime();
				return ;
			}
			
			Long serialNoZY = 0L;
			String branchNo="";
			String ldate=""; 
			
			//委托数据N秒前遗漏补推
			String nowDate = DateTime.now().toString("yyyyMMdd");
			SELECT_SQL_ENTRUST = "SELECT 'gw_entrust' as message_type ,'8815' as branch_no, aa.* FROM "+entrustTable+" aa  where  ordertype not like '%C%' and  offertime<?  and not exists(select 1 from "+entrustKafkaTable+" bb where bb.branch_no='8815' and aa.serialnum=bb.serial_no)"; //bb.l_date='"+nowDate+"'
			INSERT_SQL_ENTRUST="insert into "+entrustKafkaTable+"(l_date,branch_no,serial_no) values(?,?,?)";
			List<Map<String,Object>> dataList = jdbcTemplate.queryForList(SELECT_SQL_ENTRUST,DateTime.now().minusSeconds(120).toString("yyyyMMddHHmmss"));
			MessageTransform tran=new MessageTransform();
			List<Object[]> insertList=new ArrayList<Object[]>();
			for(Map<String,Object> data : dataList){
				String stock_account = data.get("regid").toString();
				ResponseMessage message =tran.responseTransform(data) ;
				//template.send(entrustTopic, stock_account, mapper.writeValueAsString(message));
				logger.debug("resend:"+mapper.writeValueAsString(message));
				serialNoZY = ((BigDecimal)data.get("serialnum")).longValue();
				branchNo=data.get("branch_no").toString();
				ldate=data.get("offertime").toString().substring(0,8);
				insertList.add(new Object[]{ldate,branchNo,serialNoZY}); 
			} 
			
			if(!insertList.isEmpty()){
				  jdbcTemplate.batchUpdate(INSERT_SQL_ENTRUST, insertList);

			 } 
			
			//成交数据N秒前遗漏补推
			SELECT_SQL_TRADE = "SELECT 'am trade' as message_type ,'8815' as branch_no, aa.* FROM "+tradeTable+" aa  where knocktime<?  and not exists(select 1 from "+tradeKafkaTable+" bb where bb.l_date='"+nowDate+"' and bb.branch_no='8815' and aa.serialnum=bb.serial_no)";
			INSERT_SQL_TRADE="insert into "+tradeKafkaTable+"(l_date,branch_no,serial_no) values(?,?,?)";
			insertList.clear();
		    dataList = jdbcTemplate.queryForList(SELECT_SQL_TRADE,DateTime.now().minusSeconds(120).toString("yyyyMMddHHmmss"));
			for(Map<String,Object> data : dataList){
				String stock_account = data.get("regid").toString();
				RequestMessage message =tran.requestTransform(data);
				//template.send(tradeTopic, stock_account, mapper.writeValueAsString(message));
				logger.debug("resend:"+mapper.writeValueAsString(message));
				serialNoZY = ((BigDecimal)data.get("serialnum")).longValue();
				branchNo=data.get("branch_no").toString();
				ldate=data.get("knocktime").toString().substring(0,8);
				insertList.add(new Object[]{ldate,branchNo,serialNoZY}); 
			}
			if(!insertList.isEmpty()){
				  jdbcTemplate.batchUpdate(INSERT_SQL_TRADE, insertList);
			 }
			
			context.setRuningTime(); 
			
		}catch(Exception e){
			logger.error(e.getMessage(),e);
		}finally{
		}
	}
	
	
}
